package homework2.miningPopularSong.program;

//import java.io.BufferedReader;
//import java.io.InputStreamReader;

import homework2.miningPopularSong.mapper.MapFileJoinRound2;
import homework2.miningPopularSong.mapper.MapFileSongsListenedRound1;
import homework2.miningPopularSong.mapper.MapFileUserProfileRound1;
import homework2.miningPopularSong.reducer.ReduceRound1;
import homework2.miningPopularSong.reducer.ReduceRound2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MiningPopularSong extends Configured implements Tool {
	
	public final static String KSONGS = "KSONGS";
    public final static String LCOUNTRY = "LCOUNTRY";
    
	static int printUsage() {
		System.out.println("\nminingPopularSong commands: "
                + "\nminingPopularSong.jar /LFM/users /LFM/plays /LFM/pop firstKsongsInChart country1,...,countryN\n"
                + "\n   Description:\n"
                + "             - /LFM/users = path in HDFS of userid-profile\n"
                + "             - /LFM/plays = path in HDFS of userid-timestamp-artid-artname-traid-traname\n"
                + "             - /LFM/pop = path in HDFS of output directory\n"
                + "             - firstKsongsInChart = number of songs in chart to show\n"
                + "             - country1,...,country = list of countries, separate from ',' , of which show the chart\n"
                + "                                      don't worry about spaces in the name of countries\n"
                + "				      the first letter of the name of countries must be upper\n");
		return -1;
	}

	public static void main(String[] args) throws Exception {
		
		/* check input */
        if (args.length < 5) {
                System.exit(printUsage());
        }

		Configuration conf = new Configuration();
		
		/* configure from keyboard the number of MapReduce tasks */
        /*InputStreamReader reader = new InputStreamReader(System.in);
        BufferedReader myInput = new BufferedReader(reader);
        String mapTasks= new String();
        String reduceTasks= new String();
        try {
                System.out.println("\nConfigure the number of MapReduce tasks\n"
                                + "NOTE: if you want the default values, ​​insert 8 in both cases");
                System.out.print("Insert number of map tasks: ");
                mapTasks = myInput.readLine();
                conf.setInt("mapreduce.job.maps", Integer.parseInt(mapTasks));
                System.out.println ("number entered: " + conf.get("mapreduce.job.maps"));
                System.out.print("Insert number of reduce tasks: ");
                reduceTasks = myInput.readLine();
                conf.setInt("mapreduce.job.reduces", Integer.parseInt(reduceTasks));
                System.out.println ("number entered: " + conf.get("mapreduce.job.reduces") + "\n");    
        }
        catch (NumberFormatException except) {
                System.out.println("ERROR: Integer expected instead of other values");
                System.exit(-1);
        }*/
		
		MiningPopularSong mps = new MiningPopularSong();
		mps.setConf(conf);

		int res = ToolRunner.run(mps, args);
		System.exit(res);
	}

	@Override
	public int run(String[] args) throws Exception {

		Configuration conf1 = this.getConf();
		Configuration conf2 = new Configuration();
		FileSystem fs = FileSystem.get(conf1);
		
		/* First round */

		Path input1 = new Path(args[0]);				// user file
		Path input2 = new Path(args[1]);				// playlist file
		Path outputTemp = new Path("/LFM/temp");        // directory for temporary output
		
		/* don't worry about spaces in the name of countries and the first letter must be upper */
		if(args.length>5){
			for(int i=5;i<args.length;i++){
				args[4]=args[4] + " " + args[i];
			}
		}
		
		conf1.setStrings(LCOUNTRY, args[4]);			// list of countries
		
		conf1.setInt("dfs.blocksize", 134217728);		// default dfs block size
		
		/* compress the map output */
		conf1.setBoolean("mapreduce.map.output.compress", true);
		conf1.setStrings("mapreduce.map.output.compression.codec", "org.apache.hadoop.io.compress.SnappyCodec");
		//conf1.setBoolean("mapreduce.output.fileoutputformat.compress", true);
		//conf1.setStrings("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.Lz4Codec");
		
		/* set number of MapTasks and ReduceTasks for job1 */
		conf1.setInt("mapreduce.job.maps", 20);
		conf1.setInt("mapreduce.job.reduces", 5);

		Job job1 = Job.getInstance(conf1);
		job1.setJarByClass(MiningPopularSong.class);
		job1.setJobName("MergeTwoInput");
		
		/* job1: mapTasks */
		MultipleInputs.addInputPath(job1, input1, TextInputFormat.class, MapFileUserProfileRound1.class);
		MultipleInputs.addInputPath(job1, input2, TextInputFormat.class, MapFileSongsListenedRound1.class);
		
		/* job1: reduceTask */
		FileOutputFormat.setOutputPath(job1, outputTemp);
		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(Text.class);
		job1.setReducerClass(ReduceRound1.class);

		job1.waitForCompletion(true);
		
		/* Second round */
       
        Path inputTemp = new Path("/LFM/temp/part*");
		Path outputFinal = new Path(args[2]);
        /* clean FileSystem: if exists, delete directory of outputFinal */
        if (fs.exists(outputFinal))
        	fs.delete(outputFinal, true);
        conf2.setInt(KSONGS, new Integer(args[3]));		// first K songs in chart
        
        /* compress the map output */
		conf2.setBoolean("mapreduce.map.output.compress", true);
		conf2.setStrings("mapreduce.map.output.compression.codec", "org.apache.hadoop.io.compress.SnappyCodec");
		
        /* set number of MapTasks and ReduceTasks for job2 */
        conf2.setInt("mapreduce.job.maps", 5);
		conf2.setInt("mapreduce.job.reduces", 1);
		
		Job job2 = Job.getInstance(conf2);
        job2.setJarByClass(MiningPopularSong.class);
        job2.setJobName("TopKsongs");
       
        /* job2: mapTask */
        FileInputFormat.setInputPaths(job2, inputTemp);
        job2.setInputFormatClass(TextInputFormat.class);
        job2.setMapperClass(MapFileJoinRound2.class);

        /* job2: reduceTask */
        FileOutputFormat.setOutputPath(job2, outputFinal);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);
        job2.setReducerClass(ReduceRound2.class);
        
        job2.waitForCompletion(true);
        
        /* clean FileSystem: if exists, delete directory of temporary file */
        if (fs.exists(outputTemp))
        	fs.delete(outputTemp, true);

		return job2.waitForCompletion(true) ? 0 : -1;
	}

}
